package com.etlmaster.rest.server;

import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;
import javax.ws.rs.QueryParam;
import javax.ws.rs.core.MediaType;

import com.etlmaster.executor.agent.Agents;
import com.etlmaster.executor.scheduler.Proc;
import com.etlmaster.executor.scheduler.ScheduleServer;
import com.etlmaster.executor.scheduler.ScheduleStatus;
import com.etlmaster.executor.scheduler.utils.CallableResult;
import com.etlmaster.executor.scheduler.utils.LogClusterInfoUtil;
import com.etlmaster.executor.scheduler.utils.SubmitResult;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.DpExecutorUtils;
import com.etlmaster.executor.utils.LogWriter;

/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@Path("etlmaster")
public class ServerRests {
	@GET
    @Path("startjob")
    @Produces(MediaType.TEXT_PLAIN)
    public String runProc(@QueryParam("jobjson") String jobjson) throws Exception {
		LogWriter.addLog("INFO","收到rest 请求 runProc:{}",jobjson);
		Map<String,Object> params = DpExecutorUtils.parseArgs(jobjson);
		String jobName=DataCollectionUtils.getPropertyFromMap(params, "f");
		String taskid=DataCollectionUtils.getPropertyFromMap(params, "t");
		if(taskid.equalsIgnoreCase("null")){
			taskid=DataCollectionUtils.getCurrentTimestampByFormate("yyyyMMdd");
		}
		Boolean concurrencyRun=Boolean.parseBoolean(DataCollectionUtils.getPropertyFromMap(params, "concurrencyRun"));
		String agentId=DataCollectionUtils.getPropertyFromMap(params, "agentId");
		
		Iterator it=params.entrySet().iterator();
		StringBuffer sb=new StringBuffer();
		while(it.hasNext())
		{
			Map.Entry<String, String> entry=(Entry<String, String>) it.next();
			if(entry.getKey().equalsIgnoreCase("f")||entry.getKey().equalsIgnoreCase("f")||entry.getKey().equalsIgnoreCase("f")){
				
			}else{
				sb.append(" -"+entry.getKey()+" "+entry.getValue());
			}
		}
		Proc proc=new Proc(jobName,taskid,concurrencyRun,sb.toString(),agentId);

		SubmitResult future =  ScheduleServer.submitTask(proc);
		if (future != null) {
			if (future.getResult() != null) {
				return "0";
			} else if (future.getExecuteResult() == ScheduleStatus.conditionNotMeet) {
				return ScheduleStatus.conditionNotMeet+"";
			}
		}
		return ScheduleStatus.nosuchAgent+"";
    }
	@GET
    @Path("startjobWithBlock")
    @Produces(MediaType.TEXT_PLAIN)
    public String runProcWithBlock(@QueryParam("jobjson") String jobjson) throws Exception {
		LogWriter.addLog("INFO","收到rest 请求 runProc:{}",jobjson);
		Map<String,Object> params = DpExecutorUtils.parseArgs(jobjson);
		String jobName=DataCollectionUtils.getPropertyFromMap(params, "f");
		String taskid=DataCollectionUtils.getPropertyFromMap(params, "t");
		if(taskid.equalsIgnoreCase("null")){
			taskid=DataCollectionUtils.getCurrentTimestampByFormate("yyyyMMdd");
		}
		Boolean concurrencyRun=Boolean.parseBoolean(DataCollectionUtils.getPropertyFromMap(params, "concurrencyRun"));
		String agentId=DataCollectionUtils.getPropertyFromMap(params, "agentId");
		
		Iterator it=params.entrySet().iterator();
		StringBuffer sb=new StringBuffer();
		while(it.hasNext())
		{
			Map.Entry<String, String> entry=(Entry<String, String>) it.next();
			if(entry.getKey().equalsIgnoreCase("f")||entry.getKey().equalsIgnoreCase("f")||entry.getKey().equalsIgnoreCase("f")){
				
			}else{
				sb.append(" -"+entry.getKey()+" "+entry.getValue());
			}
		}
		Proc proc=new Proc(jobName,taskid,concurrencyRun,sb.toString(),agentId);

		SubmitResult future =  ScheduleServer.submitTask(proc);
		if (future != null) {
			if (future.getResult() != null) {

				CallableResult result = (CallableResult) future.getResult()
						.get();
				
				return result.getResult()+"";
			} else if (future.getExecuteResult() == ScheduleStatus.conditionNotMeet) {
				return ScheduleStatus.conditionNotMeet+"";
			}
		}
		return ScheduleStatus.nosuchAgent+"";
		
    }
	
	@GET
    @Path("stopAgent")
    @Produces(MediaType.TEXT_PLAIN)
    public String stopAgent(@QueryParam("agentId") String agentId) throws Exception {
		LogWriter.addLog("INFO","收到rest 请求 stopAgent:{}",agentId);

		Agents agent=ScheduleServer.agents.get(agentId);
		if(agent!=null){
			agent.setStatus(ScheduleStatus.agentInvalid);
			LogClusterInfoUtil.updateHostStatus(agentId, -1);
			ScheduleServer.agents.remove(agentId);
			agent.stop();
		}
		return "Agent stoped";
    }
	@GET
    @Path("getAgentsTaskList")
    @Produces(MediaType.TEXT_PLAIN)
    public String getAgentTaskList() throws Exception {
		LogWriter.addLog("INFO","收到rest 请求 getAgentsTaskList");
		Iterator it=ScheduleServer.agents.keySet().iterator();
		StringBuffer sb=new StringBuffer();
		sb.append("{");
		while(it.hasNext()){
			String key=it.next().toString();
			sb.append("\""+key+"\":[");

			Agents agent=ScheduleServer.agents.get(key);
			List<Proc> procList=agent.getRunningProcs();
			for(int i=0;i<procList.size();i++){
				Proc proc=procList.get(i);
				sb.append("\""+proc.getRunParam()+"\"");
				if(i!=procList.size()-1){
					sb.append(",");
				}
			}
			sb.append("],");
		}
		sb.append("}");
		sb.deleteCharAt(sb.toString().length()-2);
		return sb.toString();
    }

}
